package com.batchengine.manager;

import com.batchengine.Constants;
import com.batchengine.pool.WorkerPool;
import com.batchengine.util.LoggingFactory;
import org.apache.log4j.Logger;

import java.util.*;

import static com.batchengine.Constants.MILLIS_IN_A_SECOND;
import static java.lang.Thread.currentThread;
import static java.util.Collections.synchronizedSet;
import static java.util.Collections.synchronizedSortedMap;

/**
 * Created by sunil.incubator@yahoo.com.
 * User: sunil.kalva
 * Date: Jul 12, 2009
 */
public abstract class WorkerManager implements Runnable {

    private static final Logger logger = LoggingFactory.createLogger();

    /**
     * The manager thread.
     */
    protected Thread managerThread = null;

    /**
     * The name assigned to this Workflow component.
     */
    protected String name;

    /**
     * To stop the manager thread, and all pools associated with it.
     */
    protected boolean stop = false;

    /**
     * Value is true if the manager thread has stopped.
     */
    protected boolean stopped = false;

    /**
     * To check is the manager has been started.
     */
    protected boolean started = false;

    /**
     * The time this TaskEngine component was started.
     */
    protected Date startTime = null;

    /**
     * The set of system pools managed by this WorkflowEngineManager
     */
    protected Map<String, WorkerPool> workerPoolMap = synchronizedSortedMap(new TreeMap<String, WorkerPool>());

    /**
     * The default amount of time (in seconds) that the manager
     * thread will sleep for between managing the pool(s).
     */
    protected static final int DEFAULT_SCHED_INTERVAL = 15;

    /**
     * The amount of time (in seconds) that the manager
     * thread will sleep for between managing the pool(s).
     * Defaults to <code>DEFAULT_SCHED_INTERVAL</code>.
     */
    protected int schedInterval = DEFAULT_SCHED_INTERVAL;

    protected WorkerManager(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }

    /**
     * Extending classes can override this method to do their class
     * specific initialization.
     */
    protected abstract void init();

    /**
     * Starts all the system pools which in turn results in the starting of
     * all system threads.
     *
     * @return true if successful, false otherwise
     */
    protected boolean activatePools() {

        boolean retVal = false;
        try {

            //start the system pool(s)
            Set<String> strings = workerPoolMap.keySet();
            for(String string : strings) {
                WorkerPool workerPool = workerPoolMap.get(string);
                workerPool.startWorkers();
            }
            retVal = true;
        } catch(Exception e) {
            e.printStackTrace();
        }
        return retVal;
    }

    /**
     * This is invoked when the manager thread is exiting. Extending classes
     * can override this method to do their own clean up.
     */
    protected void cleanUp() {
    }

    /**
     * Initializes the system pools, starts them, and then loops until
     * asked to stop.
     * <p/>
     * In every interation of the loop, the <code>WorkflowThreadPool.manage()</code>
     * method is called for each system pool, and then the thread sleeps for
     * <code>schedInterval</code> seconds.
     */
    public void run() {

        synchronized(this) {
            startTime = new Date();
            started = true;
            managerThread = currentThread();
            notifyAll();
        }

        try {

            init();
            if(activatePools()) {

                while(!stop) {

                    Set<String> strings = synchronizedSet(workerPoolMap.keySet());

                    Collection<String> deadPools = new ArrayList<String>();
                    for(String string : strings) {

                        if(null == workerPoolMap || null == workerPoolMap.get(string)) {
                            logger.error(String.format("pool : %s not found or pool empty", string));
                            continue;
                        }
                        WorkerPool workerPool = workerPoolMap.get(string);
                        if(workerPool.isAlive()) {
                            workerPool.manage();
                        } else {
                            deadPools.add(string);
                        }
                    }
                    //todo@Dont remove recurring tasks
//                    if(0 < deadPools.size()) {
//                        for (String s : deadPools) {
//                            workerPoolMap.remove(s);
//                        }
//                    }
                    try {
                        Thread.sleep(Constants.MILLIS_IN_A_SECOND * schedInterval);
                    } catch(InterruptedException e) {
                        //do nothing
                        e.printStackTrace();
                    }
                }
            }
        } catch(RuntimeException e) {
            e.printStackTrace();
            throw e;
        } finally {

            try {

                if(null == workerPoolMap) {
                    Set<String> strings = workerPoolMap.keySet();
                    for(String string : strings) {
                        WorkerPool workerPool = workerPoolMap.get(string);
                        workerPool.stopWorkers();
                    }
                }
                cleanUp();
            } finally {
                started = false;
                stop = true;
                stopped = true;
            }
        }
    }

    public synchronized void startManager() {

        if(!started) {
            if(this.name != null) {
                new Thread(this, name).start();
            } else {
                new Thread(this).start();
            }
            //'wait' until cacheServer is started (for a max of approx 5 seconds).
            for(int i = 0; !started && i < 5; i++) {
                try {
                    wait(MILLIS_IN_A_SECOND);
                } catch(InterruptedException e) {
                    //do nothing
                    e.printStackTrace();
                }
            }
        }
    }

    public synchronized void stopManager() {

        stop = true;
        managerThread.interrupt();
    }

    /**
     * Returns true if the task has stopped.
     *
     * @return a
     */
    public boolean isStopped() {
        return stopped;
    }

    public synchronized Date getStartTime() {
        return startTime;
    }

    public synchronized int getNumberOfActiveWorkers() {
        int num = 0;
        Set<String> strings = workerPoolMap.keySet();
        for(String string : strings) {
            WorkerPool workerPool = workerPoolMap.get(string);
            num += workerPool.getNumberOfWorkers();
        }
        return num;
    }

    public synchronized Vector<WorkerPool> getNumberOfActivePools() {

        Vector<WorkerPool> pools = new Vector<WorkerPool>();
        Set<String> strings = workerPoolMap.keySet();
        for(String string : strings) {
            pools.addElement(workerPoolMap.get(string));
        }
        return pools;
    }

    /**
     * Adds a system pool to this manager.
     *
     * @param workerPool the WorkflowThreadPool to ve managed by this manager.
     */
    public void addWorkerPool(WorkerPool workerPool) {
        workerPoolMap.put(workerPool.getName(), workerPool);
    }

    public void removeWorkerPool(WorkerPool workerPool) {
        if(null == workerPoolMap || null == workerPool) {
            return;
        }
        workerPoolMap.remove(workerPool.getName());
    }
}
